package cn.mjnxin.iot.base.process;

import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.mjnxin.iot.base.concurrent.ThreadFactoryImpl;

/**
 * 非阻塞命令处理器
 * @author CLG 2023年11月10日
 * @version V1.0.0
 */
public class MultiProccessWorker{
	private static Logger logger = LoggerFactory.getLogger(MultiProccessWorker.class);
	private static Map<String, MultiProcessor> processorMap = new ConcurrentSkipListMap<>();
	
	private static volatile boolean active = true; 
	
	public static void stopWorker() {
		active = false;
	}
	
	public static synchronized String initProcessor(String commandKey) {
		if (!processorMap.containsKey(commandKey)) {
			MultiProcessor processor = new MultiProcessor(1);
			processorMap.put(commandKey, processor);
			processor.start();
		}
		return commandKey;
	}

	public static void stopProcessor(String commandKey) {
		if (processorMap.containsKey(commandKey)) {
			final MultiProcessor processor = processorMap.get(commandKey);
			logger.info("停止" + commandKey + " 处理队列");
			processor.stopWorker();
			try {
				Thread.sleep(20L);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			if (processor.isShutdown()) {
				logger.info("停止" + commandKey + " 处理队列OK");
				processorMap.remove(commandKey);
			}
		}
	}
	
	public static void addCommandProcessor(ICommandProcessor commandProcessor) {
		String key = initProcessor(commandProcessor.getKey());
		try {
			processorMap.get(key).commandProcessorQueue.put(commandProcessor);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public static boolean isStoped() {
		for (MultiProcessor processor : processorMap.values()) {
			if (!processor.isShutdown()){
				return false;
			}
		}
		
		return true;
	}
	
	private static class MultiProcessor implements Runnable {
		private final int             poolSize;
		private final ExecutorService exService;
		private volatile boolean      active = true; 
		
		private final BlockingQueue<ICommandProcessor> commandProcessorQueue = new LinkedBlockingQueue<>(); // Command执行队列
		
		public MultiProcessor(int poolSize) {
			this.poolSize = poolSize;
			this.exService = Executors.newFixedThreadPool(poolSize, new ThreadFactoryImpl("CommandProcessorWorker", false));
		}

		public void stopWorker() {
			active = false;
		}
		
		public void run() {
			for (int i=0 ; i < poolSize ; i++ ){
				exService.execute(new Runnable() {
					@Override
					public void run() {
						while(MultiProcessor.this.active && MultiProccessWorker.active){
							resolve();
						}
					}
				});
			}
		}
	
		private void resolve() {
			try {
				final ICommandProcessor command = commandProcessorQueue.take();
				command.process();
			} catch (Exception e) {
				System.err.println("commandProcessorQueue.take() interrupted!");
				e.printStackTrace();
			}
			if (commandProcessorQueue.isEmpty()) {
				if (!MultiProcessor.this.active || !MultiProccessWorker.active) {
					exService.shutdownNow();
				}
			}
		}

		public boolean isShutdown() {
			if (commandProcessorQueue.isEmpty()) {
				//if (!Thread.currentThread().isInterrupted()) {
				Thread.currentThread().interrupt(); // 中断take
				//}
			}
			return exService.isShutdown() && exService.isTerminated();
		}
		
		public void start() {
			new Thread(this).start();
		}
	}
}

